package com.example.mq.mqserver.dataCenter;

import com.example.mq.common.BinaryTool;
import com.example.mq.common.MqException;
import com.example.mq.mqserver.core.MSGQueue;
import com.example.mq.mqserver.core.Message;

import java.io.*;
import java.util.LinkedList;
import java.util.Scanner;

/**
 * 硬盘消息管理
 * 通过这个类管理 硬盘上的消息文件
 */
public class MessageFileManager {
    //创建一个内部类表示 queue_stat.txt 所表示的统计信息:总消息 & 有效消息
    //因为 queue_stat.txt 主要用于 MessageFileManger 内部的操作
    static public class Stat{
        public int totalCount;//表示总消息数
        public int validCount;//表示有效消息数
    }

    //初始化
    public void init() {
    }


    //获取队列目录路径
    public String getQueueDir(String queueName){
        return "./data/" + queueName;
    }

    //获取消息的 queue_data.txt 文件
    public String getQueueDataPath(String queueName){
        return getQueueDir(queueName) + "/queue_data.txt";
    }

    //获取消息的 queue_stat.txt 文件
    public String getQueueStatPath(String queueName){
        return getQueueDir(queueName) + "/queue_stat.txt";
    }

    //读取统计文件方法
    //因为 统计文件 是一个文本文件 ---- 直接使用Scanner 语句进行读取
    private Stat readStat(String queueName) throws IOException {
        Stat stat = new Stat();
        // InputStream 进入文件.获取文件数据信息
        try(InputStream inputStream = new FileInputStream(getQueueStatPath(queueName))){
            // InputStream 进入文件.获取文件数据信息
            Scanner scanner = new Scanner(inputStream);
            stat.totalCount = scanner.nextInt();
            stat.validCount = scanner.nextInt();
            return stat;
        } catch (FileNotFoundException e) {
            e.printStackTrace();
        }
        //存在异常返回null
        return null;
    }

    //往 统计文件内 写内容
    private void writeStat(String queueName,Stat stat) throws IOException {
        // OutputStream 打开文件时 默认情况下会清空文件内容.相当于新数据覆盖旧数据
        try(OutputStream outputStream = new FileOutputStream(getQueueStatPath(queueName))){
            PrintWriter printWriter = new PrintWriter(outputStream);
            printWriter.write(stat.totalCount+"\t"+stat.validCount);
            // printWriter.flush() 将数据从缓冲区刷新到硬盘当中
            printWriter.flush();
        } catch (FileNotFoundException e) {
            e.printStackTrace();
        }
    }

    //创建队列目录以及相关文件
    public void createQueueFiles(String queueName) throws IOException {
        //获取目录
        File baseDir = new File(getQueueDir(queueName));
        if(!baseDir.exists()){
            boolean ok = baseDir.mkdirs();
            if(!ok){
                throw new IOException("创建目录失败! baseDir = "+baseDir.getAbsolutePath());
            }
        }

        //获取消息数据文件
        File queueDataFile = new File(getQueueDataPath(queueName));
        if(!queueDataFile.exists()){
            boolean ok = queueDataFile.createNewFile();
            if(!ok){
                throw new IOException("创建消息数据文件失败! queueDataFile = "+queueDataFile.getAbsolutePath());
            }
        }

        //获取消息统计数据文件
        File queueStatFile = new File(getQueueStatPath(queueName));
        if(!queueStatFile.exists()){
            boolean ok = queueStatFile.createNewFile();
            if(!ok){
                throw new IOException("创建消息数据文件失败! queueStatFile = "+queueStatFile.getAbsolutePath());
            }
        }

        //给统计数据文件设置默认值 0\t0
        Stat stat = new Stat();
        stat.totalCount = 0;
        stat.validCount = 0;
        writeStat(queueName,stat);
    }

    //删除队列目录和文件
    public void destroyQueueFiles(String queueName) throws IOException {
        //先删除 数据文件 和 统计文件
        File queueDataFile = new File(getQueueDataPath(queueName));
        boolean ok1 = queueDataFile.delete();

        File queueStatFile = new File(getQueueStatPath(queueName));
        boolean ok2 = queueStatFile.delete();
        File queueDir = new File(getQueueDir(queueName));
        boolean ok3 = queueDir.delete();
        if(!ok1 || !ok2 || !ok3){
            throw new IOException("删除队列目录和文件失败! queueDir = "+queueDir.getAbsolutePath());
        }
    }


    //检验队列文件是否存在 ---- 用于后续进行消息的持久化保存时进行校验判断
    public boolean checkFilesExits(String queueName){
        //先判断 数据文件 和 统计文件 是否存在
        File queueDataFile = new File(getQueueDataPath(queueName));
        if(!queueDataFile.exists()){
            return false;
        }
        File queueStatFile = new File(getQueueStatPath(queueName));
        if(!queueStatFile.exists()){
            return false;
        }
        return true;
    }

    //将消息放进队列之中
    public void sendMessage(MSGQueue queue, Message message) throws MqException, IOException {
        //判断当前队列的数据文件是否存在
        if(!checkFilesExits(queue.getName())){
            throw new MqException("[MessageFileManger] 队列对应的文件不存在 queueName = "+queue.getName());
        }

        //将对象进行序列化 转换成二进制的字节数组
        byte[] messageBinary = BinaryTool.toBytes(message);

        //避免 多线程对同一个队列里不同消息同时进行写入文件产生的线程安全情况 而加锁
        // 消息1 未设置offsetStar 值时,消息2就进来了,导致消息1的 offset 值异常
        // 只有对同一个队列进行操作的时候才会加锁 所以这里的加锁对象为 queue
        synchronized (queue){
            //往数据文件里写入消息. 先获取当前队列数据文件的长度 然后计算出 offsetStar & offsetEnd
            // offsetStar =  文件长度 + 4;
            // offsetEnd = 文件长度 + message 数据长度 + 4
            File queueDataFile = new File(getQueueDataPath(queue.getName()));
            message.setOffsetBeg(queueDataFile.length() + 4);
            message.setOffsetEnd(queueDataFile.length() + 4 + messageBinary.length);

            //往数据文件进行写操作 进行的是追加写入 而非是覆盖写入
            try(OutputStream outputStream = new FileOutputStream(queueDataFile,true)){
                // outputStream.write(int) --- 虽然是写入一个int 但只会写入一个字节
                // 使用 DataOutputStream
                try(DataOutputStream dataOutputStream = new DataOutputStream(outputStream)){
                    //先写消息的长度 --- 占据四个字节
                    dataOutputStream.writeInt(messageBinary.length);
                    //写入消息本身
                    dataOutputStream.write(messageBinary);
                }
            }
            //更新统计文件消息]
            Stat stat = readStat(queue.getName());
            stat.totalCount += 1;
            stat.validCount += 1;
            writeStat(queue.getName(),stat);
        }
    }

    //逻辑删除硬盘里的消息  --- 将硬盘上的 消息的isValid 值进行修改
    //1.将消息取出后反序列化成 Message
    //2.修改 消息的isValid 值
    //3.重新写入消息
    //此处的 message 必须包含有效的 offset 值
    public void deleteMessage(MSGQueue queue, Message message) throws IOException, ClassNotFoundException {
        //避免多线程并发产生的线程安全问题
        synchronized (queue){
            try(RandomAccessFile randomAccessFile = new RandomAccessFile(getQueueDataPath(queue.getName()),"rw")){
                //1.将消息取出后反序列化成 Message
                byte[] bufferSrc = new byte[(int) (message.getOffsetEnd()-message.getOffsetBeg())];
                //移动光标到message 头部位置
                randomAccessFile.seek(message.getOffsetBeg());
                randomAccessFile.read(bufferSrc);
                //转换成Message对象
                Message diskMessage = (Message)BinaryTool.toObject(bufferSrc);

                //2.修改 消息的isValid 值
                diskMessage.setIsValid((byte) 0x0);

                //3.重新写入消息
                //移动光标到message 头部位置 --- 读取文件时光标已移动到下一消息处
                byte[] bufferDest = BinaryTool.toBytes(diskMessage);
                randomAccessFile.seek(message.getOffsetBeg());
                randomAccessFile.write(bufferDest);
            }

            //改变统计数据中的有效消息个数
            Stat stat = readStat(queue.getName());
            if(stat.validCount > 0){
                stat.validCount -= 1;
            }
            writeStat(queue.getName(),stat);
        }
    }

    //使用以下方法 将读取到的消息加载到内存中
    //程序启动就会调用这个方法 将队列中的消息从硬盘获取到内存当中
    public LinkedList<Message> loadAllMessageFromQueue(String queueName) throws IOException, MqException, ClassNotFoundException {
        LinkedList<Message> messages = new LinkedList<>();
        try(InputStream inputStream = new FileInputStream(getQueueDataPath(queueName))){
            try(DataInputStream dataInputStream = new DataInputStream(inputStream)){
                //设置一个文件的 offset 值,便于设置内存中的 Message 的offset值
                long currentOffset = 0;
                //一个文件包含很多信息
                while (true){
                    //获取当前消息的长度
                    int messageSize = dataInputStream.readInt();
                    //根据长度读取消息
                    byte[] buffer = new byte[messageSize];
                    //获取实际消息的长度
                    //以下语句执行完毕 并且到达了文件尾,没有其他数据了,就会抛出 EOF异常
                    //此时整个循环就会停止
                    int actualSize = dataInputStream.read(buffer);
                    if(messageSize != actualSize){
                        //如果不匹配 说明文件有问题
                        throw new MqException("[MessageFileManger] 文件格式错误! queueName = " + queueName);
                    }
                    //将读取到的数据进行反序列化
                    Message message = (Message) BinaryTool.toObject(buffer);
                    //先判定当前消息是不是无效对象
                    if(message.getIsValid() != 0x1){
                        //对于无效消息
                        currentOffset += (4 + messageSize);
                        continue;
                    }
                    //对于有效消息
                    //设置offset值
                    message.setOffsetBeg(currentOffset + 4);
                    message.setOffsetEnd(currentOffset + 4 + messageSize);
                    //更新文件总体的 offset 值
                    currentOffset += (4 + messageSize);
                    messages.add(message);
                }
            }catch (EOFException e){
                //此处并不是“捕获”异常 而是进行业务逻辑的处理 --- 读取到文件尾并且没有其他数据就会抛出EOF异常
                System.out.println("[MessageFileManger] 恢复 Message 数据成功");
            }
        }
        return messages;
    }

    //检查当前是否需要进行 GC
    public boolean checkIsGC(String queueName) throws IOException {
        //获取 Stat 对象,利用 totalCount validCount 判断是否要GC
        Stat stat = readStat(queueName);
        if(stat.totalCount > 2000 && (double)(stat.validCount/stat.totalCount)<0.5){
            return true;
        }
        return false;
    }

    //获取GC之后 存放复制文件的新文件所在的位置
    private String getQueueDataNewPath(String queueName){
        return getQueueDir(queueName) + "/queue_data_new.txt";
    }

    //进行垃圾回收 清理掉无用的消息
    /*
    1.创建一个 新文件 queue_data_new.txt
    2.将旧文件中的有效消息迁移到 新文件
    3.将旧文件删除 新文件改名为 queue_data.txt
    4.更新 Stat的 总消息数 和 无效消息数
     */
    public void gc(MSGQueue queue) throws MqException, IOException, ClassNotFoundException {
        //针对 队列的 queue_data.txt 文件进行删除操作
        synchronized (queue){
            //计算耗时 以便出现性能问题能够及时找到 ---- 因为 垃圾回收是一个比较重量的操作
            long gcBeg = System.currentTimeMillis();
            //先查看 新文件是否存在
            //存在代表之前的GC可能进行到一半就意外终止
            File queueDataNewFile = new File(getQueueDataNewPath(queue.getName()));
            if (queueDataNewFile.exists()){
                throw new MqException("[MessageFileManger] 已存在垃圾回收复制文件.queueName ="+queue.getName());
            }
            boolean ok = queueDataNewFile.createNewFile();
            //创建文件失败
            if(!ok){
                throw new MqException("[MessageFileManger] 创建垃圾回收复制文件失败.queueDataNewFile="+
                        queueDataNewFile.getAbsolutePath());
            }

            //获取队列里所有的消息
            LinkedList<Message> messages = loadAllMessageFromQueue(queue.getName());
            //将旧文件的数据迁移到新文件
            try(OutputStream outputStream = new FileOutputStream(queueDataNewFile)){
                try(DataOutputStream dataOutputStream = new DataOutputStream(outputStream)){
                    for(Message message : messages){
                        byte[] buffer = BinaryTool.toBytes(message);
                        //先写数据的长度
                        dataOutputStream.writeInt(buffer.length);
                        dataOutputStream.write(buffer);
                    }
                }
            }
            //删除旧的文件 并将新文件进行改名
            File queueDataOldFile = new File(getQueueDataPath(queue.getName()));
            ok = queueDataOldFile.delete();
            //删除失败的情况
            if(!ok){
                throw new MqException("[MessageFileManger] 删除旧数据文件失败.queueDataOldFile = "+ queueDataOldFile.getAbsolutePath());
            }
            //将新文件进行改名
            ok = queueDataNewFile.renameTo(queueDataOldFile);
            if(!ok){
                throw new MqException("[MessageFileManger] 新数据文件重命名失败. queueDataNewFile= "+queueDataNewFile.getAbsolutePath()
                        +",queueDataOldFile = "+ queueDataOldFile.getAbsolutePath());
            }

            //修改统计文件
            Stat stat = readStat(queue.getName());
            //获取的都有有效的消息
            stat.totalCount = messages.size();
            stat.validCount = messages.size();
            writeStat(queue.getName(),stat);

            long gcEnd = System.currentTimeMillis();
            System.out.println("[MessageFileManger] GC执行完毕. queueName = "+queue.getName()+",wasteTime = "+(gcEnd-gcBeg)+"ms");
        }
    }
}
